package com.intct.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import com.intct.func.LookupAsyncDimFun;
import com.intct.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

/**
 * @author gufg
 * @since 2025-09-22 15:49
 */
public class DwdVehicleAPI {
    public static void main(String[] args) throws Exception {
        // 配置环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 开启检查
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        // 从工程中resoucres路径读取配置文件
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(DwdVehicleAPI.class.getResource("/intct.properties").getFile());

        // 数据源配置 ods kafka
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(parameterTool, "ods_vehicle");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dwd_locus_source");

        /*
            {
                "op": "r",
                "name": "vehicle_distribution",
                "after": {
                    "actionTime": 1753773834,
                    "areaId": "110119",
                    "driverId": "870",
                    "carShelfNumber": "Y51637129D",
                    "car": "24",
                    "type": "0",
                    "speed": 16
                },
                "ts_ms": 1753773834880
            }
         */
        // 1 过滤掉不在线车
        SingleOutputStreamOperator<String> filterDS =
                sourceDS.filter(value -> "0".equals(JSON.parseObject(value).getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY).getString("type")));

        // 数据转换
        SingleOutputStreamOperator<String> mapDS = filterDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String jsonStr) throws Exception {
                // 转换类型
                JSONObject jsonObj = JSON.parseObject(jsonStr);
                return jsonObj.getJSONObject(Constant.FLINK_CDC_MYSQL_JSON_AFTER_KEY).toJSONString();
            }
        });
        // TODO 2 规范
		// TODO 2.1 重复数据

		// TODO 2.2 空值

		// TODO 2.3 脱敏

		// TODO 2.4 单位

		// 3 退维
        /*
{
                    "actionTime": 1753773834,
                    "areaId": "110119",
                    "driverId": "870",
                    "carShelfNumber": "Y51637129D",
                    "car": "24",
                    "type": "0",
                    "speed": 16
                }
         */
        SingleOutputStreamOperator<String> unorderedWait =
                AsyncDataStream.unorderedWait(
                        mapDS,
                        new LookupAsyncDimFun("intct84", "dim_area_info", "areaId"),
                        30,
                        TimeUnit.SECONDS,
                        1000);

        unorderedWait.print();
		// 4 写在Kafka
        unorderedWait.sinkTo(KafkaUtil.getKafaSink(parameterTool, "dwd_vehicle", "dwd_vehicle_"));

        // 启动作业
        env.execute("dwd_vehicle_job");
    }
}
